package top.yehot.kettle.boot.core.utils;

import java.io.File;
import java.util.Optional;
import java.util.Properties;
import java.util.function.BiConsumer;

import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.RepositoriesMeta;
import org.pentaho.di.repository.RepositoryMeta;
import org.pentaho.di.repository.filerep.KettleFileRepository;
import org.pentaho.di.repository.filerep.KettleFileRepositoryMeta;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.springframework.util.StringUtils;

import lombok.extern.slf4j.Slf4j;
import top.yehot.cmm.util.core.base.BaseException;
import top.yehot.cmm.util.spring.utils.SpringUtil;
import top.yehot.kettle.boot.core.SysConst;
import top.yehot.kettle.boot.enums.KettleResEnum;
import top.yehot.kettle.boot.models.entity.Repository;
import top.yehot.kettle.boot.service.RepositoryService;

/**
 * Kettle资源库工具类
 * 
 * @author lquan
 * @date 2022年5月29日 下午4:04:07
 */
@Slf4j
public class RepositoryUtil {

	/**
	 * 判断资源库是否连接成功
	 *
	 * @param repId 资源库ID
	 * @return {@link Boolean}
	 */
	public static synchronized boolean isExistCache(String repId) {
		return SysConst.REP_DATABASE.containsKey(repId);
	}
	
	/**
	 * 删除缓存中的资源库
	 */
	public static synchronized void removeRepositoryCache(String repId) {
		SysConst.REP_DATABASE.remove(repId);
	}

	/**
	 * 通过资源库ID获取资源库(缓存)
	 *
	 * @param repId 资源库ID
	 * @return {@link AbstractRepository}
	 */
	public static synchronized AbstractRepository getRepositoryCache(String repId) {
		return SysConst.REP_DATABASE.getOrDefault(repId, null);
	}

	/**
	 * 通过资源库ID获取资源库
	 *
	 * @param repId 资源库id
	 * @return {@link AbstractRepository}
	 */
	public static synchronized AbstractRepository getRepository(String repId) {
		if (!StringUtils.hasText(repId)) return null;
		
		// 判断资源库连接是否存在，不存在创建
		if (!isExistCache(repId)) {
			return createConnection(getRepositoryInfo(repId));
		}
		
		AbstractRepository abstractRepository = getRepositoryCache(repId);
		// 判断资源库是否已断开连接，如果断开连接就重连
		if (!abstractRepository.isConnected()) {
			removeRepositoryCache(repId);
			return createConnection(getRepositoryInfo(repId));
		}
		
		// 如果是数据库资源库需要判断数据库连接是否已断开连接，断开就要重连
		if (abstractRepository instanceof KettleDatabaseRepository) {
			KettleDatabaseRepository dbRepo = (KettleDatabaseRepository) abstractRepository;
			try {
				if (dbRepo.getDatabase().getConnection().isClosed()) {
					abstractRepository.init(abstractRepository.getRepositoryMeta());
					abstractRepository.disconnect();
					
					Repository repositoryInfo = getRepositoryInfo(repId);
					abstractRepository.connect(repositoryInfo.getRepUsername(), repositoryInfo.getRepPassword());
				}
			} catch (Exception e) {
				log.error(KettleResEnum.RESTART_CONNECTION_ERR.getMsg() + "，repId:{}", repId);
				throw new BaseException(KettleResEnum.RESTART_CONNECTION_ERR);
			}
		}
		return abstractRepository;
	}

	/**
	 * 创建资源库链接
	 *
	 * @param repo 连接参数
	 */
	public static synchronized AbstractRepository createConnection(Repository repo) {
		// 不存在就创建资源库
		AbstractRepository repository = null;
		switch (repo.getRepType()) {
		case file:
			repository = fileRepository(repo);
			break;
		case db:
			repository = databaseRepository(repo);
			break;
		default:
			throw new BaseException("Unexpected value: " + repo.getRepType());
		}
		return repository;
	}

	/**
	 * 更新 repositories.xml
	 * 
	 * @param repId 资源库ID
	 */
	public static synchronized void writeRepositoryXml(String repId, BiConsumer<AbstractRepository, RepositoriesMeta> bc) {
		// 获取资源库
		AbstractRepository abstractRepository = getRepository(repId);
		// 获取资源库名称
		String repName = abstractRepository.getName();
		try {
			RepositoriesMeta repositories = new RepositoriesMeta();
			if (!repositories.readData()) return;
			
			// 先删除旧的资源库
			bc.accept(abstractRepository, repositories);
			
			RepositoryMeta previousMeta = repositories.searchRepository(repName);
			if (null != previousMeta) {
				repositories.removeRepository(repositories.indexOfRepository(previousMeta));
			}
			repositories.addRepository(abstractRepository.getRepositoryMeta());
			
			if (abstractRepository instanceof KettleDatabaseRepository) {
				KettleDatabaseRepository dbRepo = (KettleDatabaseRepository) abstractRepository;
				DatabaseMeta dbPreviousMeta = repositories.searchDatabase(repName + SysConst.REP_DB_SUFFIX);
				if (null != previousMeta) {
					repositories.removeDatabase(repositories.indexOfDatabase(dbPreviousMeta));
				}
				repositories.addDatabase(dbRepo.getDatabaseMeta());
			}
			repositories.writeData();
		} catch (Exception e) {
			log.error(KettleResEnum.WRITE_REPOSITORY_XML_ERR.getMsg() + "，repId:{}", repId, e);
			throw new BaseException(KettleResEnum.WRITE_REPOSITORY_XML_ERR);
		}
	}
	
	/**
	 * 删除 repositories.xml 文件节点
	 * 
	 * @param repId 资源库ID
	 */
	public static synchronized void delRepositoryXml(String repName) {
		try {
			RepositoriesMeta repositories = new RepositoriesMeta();
			if (!repositories.readData())
				return;
			
			// 查询是否有同名资源库没有就新建
			RepositoryMeta previousMeta = repositories.searchRepository(repName);
			if (null != previousMeta) {
				repositories.removeRepository(repositories.indexOfRepository(previousMeta));
			}
			DatabaseMeta dbPreviousMeta = repositories.searchDatabase(repName + SysConst.REP_DB_SUFFIX);
			if (null != previousMeta) {
				repositories.removeDatabase(repositories.indexOfDatabase(dbPreviousMeta));
			}
			repositories.writeData();
		} catch (Exception e) {
			log.error(KettleResEnum.WRITE_REPOSITORY_XML_ERR.getMsg() + "，repName:{}", repName, e);
			throw new BaseException(KettleResEnum.WRITE_REPOSITORY_XML_ERR);
		}
	}

	/**
	 * 连接到文件资源库
	 *
	 * @param fileRep 连接参数
	 */
	private static AbstractRepository fileRepository(Repository fileRep) {
		// 判断文件是否存在
		String baseDir = FileUtil.replaceSeparator(fileRep.getRepBasePath());
		if (StringUtils.hasText(baseDir)) {
			File file = new File(baseDir);
			if (!file.exists() || !file.isDirectory()) {
				throw new BaseException(KettleResEnum.FILE_REPOSITORY_NOT_EXIST);
			}
		} else {
			throw new BaseException(KettleResEnum.FILE_REPOSITORY_NOT_EXIST);
		}

		// 文件资源库元数据
		KettleFileRepositoryMeta frm = new KettleFileRepositoryMeta();
		frm.setBaseDirectory(baseDir);
		frm.setName(fileRep.getRepName());

		// 初始化资源库
		KettleFileRepository rep = new KettleFileRepository();
		rep.init(frm);
		// 开始连接资源库
		try {
			rep.connect(fileRep.getRepUsername(), fileRep.getRepPassword());
		} catch (KettleException e) {
			log.error(KettleResEnum.FILE_REPOSITORY_CONNECTION_ERR.getMsg(), e);
			throw new BaseException(KettleResEnum.FILE_REPOSITORY_CONNECTION_ERR);
		}

		// 缓存资源库信息
		if (StringUtils.hasText(fileRep.getId())) {
			SysConst.REP_DATABASE.put(fileRep.getId(), rep);
		}
		return rep;
	}

	/**
	 * 连接到数据库资源库
	 *
	 * @param dbRep 连接参数
	 */
	private static AbstractRepository databaseRepository(Repository dbRep) {
		// 获取数据连接元
		DatabaseMeta dataMeta = new DatabaseMeta(dbRep.getRepName() + SysConst.REP_DB_SUFFIX, dbRep.getDbType(), dbRep.getDbAccess(),
				dbRep.getDbHost(), dbRep.getDbName(), dbRep.getDbPort(), dbRep.getDbUsername(), dbRep.getDbPassword());
		// 设置连接池
//		dataMeta.setUsingConnectionPool(true);
//		dataMeta.setInitialPoolSize(20);
//		dataMeta.setMaximumPoolSize(100);
		dataMeta.setAttributes(setProperties(dataMeta.getAttributes(), dbRep.getDbType()));

		// 数据库资源库元
		KettleDatabaseRepositoryMeta drm = new KettleDatabaseRepositoryMeta();
		drm.setConnection(dataMeta);
		drm.setName(dbRep.getRepName());

		// 初始化并连接到数据库资源库
		KettleDatabaseRepository rep = new KettleDatabaseRepository();
		rep.init(drm);
		// 开始连接资源库
		try {
			rep.connect(dbRep.getRepUsername(), dbRep.getRepPassword());
		} catch (KettleException e) {
			log.error(KettleResEnum.DB_REPOSITORY_CONNECTION_ERR.getMsg(), e);
			throw new BaseException(KettleResEnum.DB_REPOSITORY_CONNECTION_ERR);
		}

		// 缓存资源库信息
		if (StringUtils.hasText(dbRep.getId())) {
			SysConst.REP_DATABASE.put(dbRep.getId(), rep);
		}
		return rep;
	}
	
	/**
	 * 查询数据库获取资源库连接信息
	 * 
	 * @param repId 资源库ID
	 * @return 资源库连接信息
	 */
	private static Repository getRepositoryInfo(String repId) {
		Repository repositoryInfo = SpringUtil.getBean(RepositoryService.class).getById(repId);
		Optional.ofNullable(repositoryInfo).orElseThrow(() -> new BaseException(KettleResEnum.REPOSITORY_NOT_EXIST));
		return repositoryInfo;
	}

	/**
	 * 设置数据库属性，同客户端数据库配置界面
	 */
	private static Properties setProperties(Properties properties, String databaseType) {
		// 设置端口号
//    	properties.setProperty("PORT_NUMBER", "3306");
		// 使用结果流游标
		properties.setProperty("STREAM_RESULTS", "");

		/** 高级配置 **/
		// 支持时间戳数据类型
//        properties.setProperty("SUPPORTS_TIMESTAMP_DATA_TYPE", "Y");
		// 支持布尔数据类型
//        properties.setProperty("SUPPORTS_BOOLEAN_DATA_TYPE", "Y");
		// 标识符使用引号括起来
		properties.setProperty("QUOTE_ALL_FIELDS", "N");
		// 强制标识符为小写
		properties.setProperty("FORCE_IDENTIFIERS_TO_LOWERCASE", "N");
		// 强制标识符大写
		properties.setProperty("FORCE_IDENTIFIERS_TO_UPPERCASE", "N");
//        properties.setProperty("PRESERVE_RESERVED_WORD_CASE", "");
		// 默认模式名，在没有其他模式名时启用
//        properties.setProperty("PREFERRED_SCHEMA_NAME", "");
		// 连接成功以后要执行的SQL，多条使用分号隔开
//        properties.setProperty("SQL_CONNECT", "");

		/** 选项 命名参数 **/
		if ("MySQL".equalsIgnoreCase(databaseType)) {
			// 最小交互数据量，如设置为10的话，查询100条数据，需要和数据库交互10次
			properties.setProperty("EXTRA_OPTION_MYSQL.defaultFetchSize", "500");
			properties.setProperty("EXTRA_OPTION_MYSQL.useCursorFetch", "true");
			// 配置字符集
			properties.setProperty("EXTRA_OPTION_MYSQL.characterEncoding", "utf8");
			// 不使用SSL连接MYSQL
			properties.setProperty("EXTRA_OPTION_MYSQL.useSSL", "false");
			// 配置断线自动重新链接
//            properties.setProperty("EXTRA_OPTION_MYSQL.autoReconnect", "true");
		}

		/** 连接池配置,布尔类型必须使用n/y，不区分大小写 **/
		// 是否使用连接池
		properties.setProperty("USE_POOLING", "Y");
		// 初始化连接数目
		properties.setProperty("POOLING_INITIAL_POOL_SIZE", "10");
		// 允许最大连接数
		properties.setProperty("POOLING_MAXIMUM_POOL_SIZE", "200");
		// 控制 PoolGuard 是否允许访问底层连接
//        properties.setProperty("POOLING_accessToUnderlyingConnectionAllowed", "");
		// 连接池创建的默认连接目录
//        properties.setProperty("POOLING_defaultCatalog", "");
		// 是否只读
//        properties.setProperty("POOLING_defaultReadOnly", "N");
		// 此池创建的连接的默认 TransactionIsolation 状态。
//        properties.setProperty("POOLING_defaultTransactionIsolation", "");
		// 启动池时创建的初始连接数。
//        properties.setProperty("POOLING_initialSize", "");
		// 是否自动提交
		properties.setProperty("POOLING_defaultAutoCommit", "Y");
		// 连接池支持的最大连接数
		properties.setProperty("POOLING_maxActive", "80");
		// 连接池中最多可空闲maxIdle个连接
		properties.setProperty("POOLING_maxIdle", "10");
		// 可以同时从语句池中分配的打开语句的最大数量，或者为零表示无限制。
		properties.setProperty("POOLING_maxOpenPreparedStatements", "0");
		// 连接池中连接用完时,新的请求等待时间,毫秒，-1，表示无限等待，直到超时为止
		properties.setProperty("POOLING_maxWait", "-1");
		// 连接池中最小空闲连接数，当连接数少于此值时，连接池会创建连接来补充到该值的数量
		properties.setProperty("POOLING_minIdle", "10");
		// 为此池启用准备好的语句池
//        properties.setProperty("POOLING_poolPreparedStatements", "");
		// 标记以在超过 removeAbandonedTimeout 时删除放弃的连接
//        properties.setProperty("POOLING_removeAbandoned", "");
		// 可以删除放弃的连接之前的超时时间（以秒为单位）。
//        properties.setProperty("POOLING_removeAbandonedTimeout", "");
		// 申请连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能。
		properties.setProperty("POOLING_testOnBorrow", "Y");
		// 归还连接时执行validationQuery检测连接是否有效，做了这个配置会降低性能
		properties.setProperty("POOLING_testOnReturn", "N");
		// 建议配置为Y，不影响性能，并且保证安全性。申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效。
		properties.setProperty("POOLING_testWhileIdle", "N");
		// 空闲对象驱逐线程运行之间休眠的毫秒数。 当为非肯定时，不会运行空闲的对象驱逐线程
//        properties.setProperty("POOLING_timeBetweenEvictionRunsMillis", "");
		// 验证数据库连接的查询语句
		properties.setProperty("POOLING_validationQuery", "select 2");

		/** 集群配置 **/
		// 是否为集群
//        properties.setProperty("IS_CLUSTERED", "N");
		return properties;
	}

}
